package com.shujia.flink.tf

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

object DemoMap {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //读取文件构建DS
    val studentsDS: DataStream[String] = env.readTextFile("flink/data/students.txt")

    //java api
    val clazzDS: DataStream[String] = studentsDS.map(new Map())

    clazzDS.print()

    //scala api
    val scalaApiDS: DataStream[String] = studentsDS.map(stu => stu.split(",")(4))


    env.execute()

  }

}

class Map extends MapFunction[String, String] {
  /**
   * map:毛一条数据执行一次时，进来一条数据返回一条数据
   *
   * @param value ：一条数据
   * @return
   */

  override def map(value: String): String = {
    //取出班级
    val clazz: String = value.split(",")(4)
    //返回数据
    clazz
  }
}
